package rxjava;

/**
 * 创建时间：2018年03月06日16:44 <br>
 * 作者：fuchaoyang <br>
 * 描述：自己实现简单的 RxJava
 */

public class Observable<T> {
  final OnSubscribe<T> onSubscribe;

  private Observable(OnSubscribe<T> onSubscribe) {
    this.onSubscribe = onSubscribe;
  }

  public static <T> Observable<T> create(OnSubscribe<T> onSubscribe) {
    return new Observable<T>(onSubscribe);
  }

  public void subscribe(Subscriber<? super T> subscriber) {
    subscriber.onStart();
    onSubscribe.call(subscriber);
  }

  public <R> Observable<R> map(final Transformer<? super T, ? extends R> transformer) {
    return create(new OnSubscribe<R>() { // 生成一个桥接的Observable和 OnSubscribe
      @Override public void call(final Subscriber<? super R> subscriber) {
        Observable.this.subscribe(new Subscriber<T>() { // 订阅上层的Observable
          @Override public void onCompleted() {
            subscriber.onCompleted();
          }

          @Override public void onError(Throwable t) {
            subscriber.onError(t);
          }

          @Override public void onNext(T var1) {
            // 将上层的onSubscribe发送过来的Event，通过转换和处理，转发给目标的subscriber
            subscriber.onNext(transformer.call(var1));
          }
        });
      }
    });
  }

  public interface Transformer<T, R> {
    R call(T from);
  }

  public interface OnSubscribe<T> {
    void call(Subscriber<? super T> subscriber);
  }
}
